package org.yonggan.dmp.etl

import java.util.Properties

import com.alibaba.fastjson.JSON
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.{SQLContext, SaveMode}
import org.apache.spark.{SparkConf, SparkContext}
import org.yonggan.dmp.conf.ConfigManager
import org.yonggan.dmp.utils.GeoHashEx

/**
  * 提取用户的经纬度
  */
object Lbs2GeoHashMySQL {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName("提取用户的经纬度")
      .setMaster("local[*]")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

//    val sc = new SparkContext(conf)
//    val sqlContext = new SQLContext(sc)
//
//    // 读取文件
//    val baseDF = sqlContext.read.parquet(ConfigManager.PARQUET_OUT)
//
//    import sqlContext.implicits._
//    val resultDF = baseDF.select("long", "lat").filter(
//      """
//        long != 0 and lat != 0
//      """.stripMargin).distinct()
//      .map(row => (row.get(0), row.get(1)))
////        .filter(f=>
////          MapApiUtils.checkIsChina(f._2.asInstanceOf[String],
////            f._1.asInstanceOf[String]) ) // 过滤掉不是中国的经纬度
//      .map(row => {
//        // 经度
//        val lnt = row._2.asInstanceOf[String]
//        //纬度
//        val lat = row._1.asInstanceOf[String]
//        // 提取商圈信息
//        val json = HttpClientUtils.execute(lnt, lat)
//        val resJson: String = MapApiUtils.baiduMapApiResp(json)
//
//        val jsonObject = JSON.parseObject(resJson)
//        val business = jsonObject.getJSONObject("result").getString("business")
//
//        (lnt, lat, business)
//      }).filter(f => StringUtils.isNotEmpty(f._3)).map(tp => (GeoHashEx.enCode(tp._2, tp._1)(12), tp._3))
//      .toDF("geoHash", "business")
//
//    val prop = new Properties()
//    prop.setProperty("user", "root")
//    prop.setProperty("password", "1234")
//    prop.setProperty("driver", "com.mysql.jdbc.Driver")
//
//    // 存入数据库
//    resultDF.write.mode(SaveMode.Append)
//      .jdbc("jdbc:mysql://localhost:3306/dmp?characterEncoding=utf8", "dmp_lbs_business", prop)
//

//    sc.stop()
  }

}
